package org.itzixi.zk;

import jakarta.annotation.Resource;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.itzixi.pojo.netty.NettyServerNode;
import org.itzixi.utils.JsonUtils;
import org.itzixi.utils.RedisOperator;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@ConfigurationProperties(prefix = "zookeeper.curator")
@Data
public class CuratorConfig {

    @Autowired
    private RedisOperator redis;
    @Resource
    private RabbitAdmin rabbitAdmin;

    public String host;                    // Zookeeper 地址
    public Integer connectionTimeoutMs;    // 连接超时时间
    public Integer sessionTimeoutMs;       // 会话超时时间
    public Integer sleepMsBetweenRetry;    // 创建会话失败时，每次重试的间隔时间
    public Integer maxRetries;             // 创建会话失败时，最大重试次数
    public String namespace;               // 命名空间

    public static final String PATH="/server-list";

    @Bean("curatorClient")
    public CuratorFramework curatorClient(){
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(sleepMsBetweenRetry, maxRetries);
        // 创建 CuratorFramework 客户端
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(host)
                .connectionTimeoutMs(connectionTimeoutMs)
                .sessionTimeoutMs(sessionTimeoutMs)
                .retryPolicy(retryPolicy)
                .namespace(namespace)
                .build();
        // 启动客户端
        client.start();

// 测试
//        try {
//            client.create()
//                    .creatingParentsIfNeeded()
//                    .forPath("/springboot/test","test".getBytes());
//        } catch (Exception e) {
//            throw new RuntimeException(e);
//        }
        addWatcher(PATH,client);
        return client;
    }

    /**
     * 添加Watcher监听器
     * @param path
     * @param client
     */
    public void addWatcher(String path, CuratorFramework client) {
        // TODO 添加Watcher监听器的实现
        CuratorCache curatorCache = CuratorCache.build(client, path);
        curatorCache.listenable().addListener((type, oldData, data) -> {
            System.out.println(type);

//            if(oldData != null){
//                System.out.println("old path = " + oldData.getPath());
//                System.out.println("old data = " + String.valueOf(oldData.getData()));
//            }
//
//            if(data != null){
//                System.out.println("new path = " + data.getPath());
//                System.out.println("new data = " + String.valueOf(data.getData()));
//            }

            switch (type.name()) {
                case "NODE_CREATED":
                    log.info("(子)节点创建");
                    break;
                case "NODE_CHANGED":
                    log.info("(子)节点(数据)变更");
                    break;
                case "NODE_DELETED":
                    log.info("(子)节点删除");

                    NettyServerNode oldNode = JsonUtils.jsonToPojo(
                            new String(oldData.getData()), NettyServerNode.class);

                    System.out.println("old path = " + oldData.getPath());
                    System.out.println("old value = " + oldNode);

                    // 移除残留端口
                    String oldPort = oldNode.getPort()+"";
                    String portKey = "netty_port";
                    redis.hdel(portKey, oldPort);

                    // 移除残留RabbitMQ队列
                    String queueName = "netty_queue_" + oldPort;
                    rabbitAdmin.deleteQueue(queueName);

                    break;
                default:
                    log.info("default...");
                    break;
            }

        });
        curatorCache.start();
    }

}
